package com.jianggy.pulsar;

import org.apache.pulsar.client.api.*;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 *第一个pulsar程序，producer
 *@author jianggy
 *@date 2022/12/18 
 *@version 1.0
 **/
public class PulsarProducer {

    public static void main(String[] args) throws Exception {
        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://192.168.10.241:16650").build();
        String npTopic = "non-persistent://public/default/jianggy";
        String subscriptionName = "jianggy-subscription";
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        executorService.submit(() -> {
            Producer<byte[]> producer = null;
            try {
                producer = pulsarClient.newProducer().topic(npTopic).create();
            } catch (PulsarClientException e) {
                e.printStackTrace();
            }
            int i = 100;
            while (i > 0) {
                try {
                    MessageId messageId = producer.send(dateFormat.format(new Date()).getBytes());
                    System.out.println(messageId.toString());
                    TimeUnit.SECONDS.sleep(1);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    i--;
                }
            }
        });
        executorService.submit(() -> {
            int i = 0;
            Consumer<byte[]> consumer = null;
            try {
                consumer = pulsarClient.newConsumer().topic(npTopic).subscriptionName(subscriptionName)
                        .subscribe();
            } catch (PulsarClientException e) {
                e.printStackTrace();
            }
            while (i < 100) {
                try {
                    Message<byte[]> message = consumer.receive();
                    System.out.println("接受消息：" + new String(message.getValue()));
                    consumer.acknowledge(message.getMessageId());
                } catch (PulsarClientException e) {
                    e.printStackTrace();
                }
            }

        });
    }
}
